package com.hqq.concurrency.reactor.framework;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * AbstractNIOChannel
 * 抽象非阻塞NIO通道类
 * 将操作系统管理的资源可以提交到{@link NIOReactor}
 * <p>
 * 功能:
 * 扮演着读写数据的角色.提供了一种阻塞式的写机制 当任何{@link ChannelHandler}想要写回数据时,将数据放入写队列
 * 具有更好的吞吐量
 * Created by heqianqian on 2017/8/5.
 */
public abstract class AbstractNIOChannel {

    private final SelectableChannel channel;

    private final ChannelHandler handler;

    private final Map<SelectableChannel, Queue<Object>> channelToPendingWrites
            = new ConcurrentHashMap<>();

    private NIOReactor reactor;

    /**
     * 创建一个新的通道
     *
     * @param channel 通道
     * @param handler 处理器
     */
    public AbstractNIOChannel(SelectableChannel channel, ChannelHandler handler) {
        this.channel = channel;
        this.handler = handler;
    }

    /**
     * 向当前channel注入reactor
     */
    void setReactor(NIOReactor reactor) {
        this.reactor = reactor;
    }

    /**
     * 返回NIO channel
     */
    public SelectableChannel getJavaChannel() {
        return channel;
    }

    /**
     * 返回当前通道要执行的操作 提供给{@link Selector}
     *
     * @return 需要进行的操作
     * @see SelectionKey
     */
    public abstract int getInterestedOps();

    /**
     * 将通道绑定到指定端口
     */
    public abstract void bind() throws IOException;

    /**
     * 根据key读取的数据
     */
    public abstract Object read(SelectionKey key) throws IOException;

    /**
     * 返回与当前通道关联的处理器handler
     *
     * @return handler
     */
    public ChannelHandler getHandler() {
        return handler;
    }

    public void flush(SelectionKey key) throws IOException {
        Queue<Object> pendingWrites = channelToPendingWrites.get(key.channel());
        while (true) {
            //需要写入的数据
            Object pendingWrite = pendingWrites.poll();
            if (pendingWrite == null) {
                //无数据可写因此可以进行读操作
                reactor.changeOps(key, SelectionKey.OP_READ);
                break;
            }
            //调用具体的channel进行写操作
            doWrite(pendingWrite, key);
        }

    }

    /**
     * 向通道中写入数据
     *
     * @param pendingWrite 数据
     */
    protected abstract void doWrite(Object pendingWrite, SelectionKey key) throws IOException;

    /**
     * 将要写入通道的数据入队 该方法并不保证方法返回时就写入数据 而是在通道被刷新时才会将数据写入
     * {@link ChannelHandler} 使用该方法给客户端发送回复请求
     *
     * @param data 要写入通道的数据
     */
    public void write(Object data, SelectionKey selectionKey) {
        Queue<Object> pendingWriteData = this.channelToPendingWrites.get(selectionKey.channel());
        if (pendingWriteData == null) {
            synchronized (this.channelToPendingWrites) {
                pendingWriteData = this.channelToPendingWrites.get(selectionKey.channel());
                if (pendingWriteData == null) {
                    pendingWriteData = new ConcurrentLinkedQueue<>();
                    channelToPendingWrites.put(selectionKey.channel(), pendingWriteData);
                }
            }
        }
        pendingWriteData.add(data);
        reactor.changeOps(selectionKey, SelectionKey.OP_WRITE);
    }
}
